package cn.com.idmy.mq.listener;

import cn.com.idmy.mq.coder.MqMsgDecoder;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;

import java.lang.reflect.Method;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class OrderlyMsgListener extends AbstractMsgListener implements MessageListenerOrderly {
    public OrderlyMsgListener(Map<String, Map<String, Class<?>>> messageTypeMap, Map<String, Map<String, Method>> methodMap, Map<String, Map<String, Object>> beanMap, Map<String, Map<String, MqMsgDecoder>> decoderMap) {
        super(messageTypeMap, methodMap, beanMap, decoderMap);
    }

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageList, ConsumeOrderlyContext context) {
        Iterator<MessageExt> iterator = messageList.iterator();
        MessageExt ext;
        do {
            if (iterator.hasNext()) {
                ext = iterator.next();
            } else {
                return ConsumeOrderlyStatus.SUCCESS;
            }
        } while (this.invoke(ext));

        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
    }
}
